package cn.doitedu.data_snyc;

import cn.doitedu.utils.GeohashUDF;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/14
 * @Desc: 学大数据，上多易教育
 *   gps参考点地理位置表，同步到 hbase
 *   维表数据是一个静态表，所以本程序可以直接以批模式运行
 **/
public class Job02_GpsArea2Hbase {

    public static void main(String[] args) {
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射mysql中地理位置参考表： tmp_area
        tenv.executeSql(
                " create table tmp_area_mysql (                       "+
                        "   lat double,                                       "+
                        " 	lng double,                                       "+
                        " 	province string,                                  "+
                        " 	city string,                                      "+
                        " 	region string                                     "+
                        " ) WITH (                                            "+
                        "    'connector' = 'jdbc',                            "+
                        "    'url' = 'jdbc:mysql://doitedu:3306/realtimedw',  "+
                        "    'table-name' = 'tmp_area',                       "+
                        "    'driver' = 'com.mysql.cj.jdbc.Driver',           "+
                        "    'username' = 'root',                             "+
                        "    'password' = 'root'                              "+
                        " );                                                  "
        );



        // 建表，映射hbase中的地理位置维表： dim_geo_area
        // hbase> create 'dim_geo_area','f'
        tenv.executeSql(
                " create table dim_geo_area_hbase(                     "+
                        "    geohash  string,                                  "+
                        "    f row<province string,city string,region string>, "+
                        "    primary key(geohash) not enforced                 "+
                        " ) WITH (                                             "+
                        "  'connector' = 'hbase-2.2',                          "+
                        "  'table-name' = 'dim_geo_area',                      "+
                        "  'zookeeper.quorum' = 'doitedu:2181'                 "+
                        " )                                                    "
        );


        // insert into select
        tenv.createTemporaryFunction("geo", GeohashUDF.class);
        tenv.executeSql(
                "insert into dim_geo_area_hbase " +
                "select " +
                "    geo(lat,lng) as geohash,  " +
                "    row(province,city,region) as f " +
                "from tmp_area_mysql"
        );


    }

}
